package net.bwie.realtime.mall.dim.sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;

import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @ClassName：DimSinkHBaseFunction
 * @Author: ZddddQ
 * @Date: 2025/6/3 13:40
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class DimSinkHBaseFunction extends RichSinkFunction<String> {

    // 定义连接
    private Connection connection = null;

    // 初始化连接
    @Override
    public void open(Configuration parameters) throws Exception {

        // 1.初始化连接
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        // 2.设置zk集群信息
        conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
        // 3.设置zk端口信息
        conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
        // 4.创建HBase连接
        connection = ConnectionFactory.createConnection(conf) ;

    }

    @Override
    public void invoke(String value, Context context) throws Exception {

        // 1.解析JSON获取数据
        JSONObject jsonObject = JSON.parseObject(value);
        String tableName = jsonObject.getString("hbase_table_name");
        String familyName = jsonObject.getString("family_name");
        String rowKeyColumn = jsonObject.getString("row_key_column");
        String operatorType = jsonObject.getString("operator_type");
        JSONObject operatorData = jsonObject.getJSONObject("operator_data");

        //System.out.println(rowKeyColumn);
        //System.out.println(jsonObject.toString());


        // 2.创建表
        createHbaseTable(tableName, familyName);

        // 3.根据表名获取HBase对象
        Table table = connection.getTable(TableName.valueOf(tableName));

        // 4.判断操作类型
        if("insert".equals(operatorType) || "update".equals(operatorType)){
            // 4.1 获取rowKey
            String rowKeyValue = operatorData.getString(rowKeyColumn);
            // 4.2 创建Put对象
            Put put = new Put(Bytes.toBytes(rowKeyValue));
            // 4.3 遍历，JSONObject获取列名和列值
            for (String columnName : operatorData.keySet()) {
                // 获取列值
                String columnValue = operatorData.getString(columnName);
                // 如果列值不为空，则将其添加到Put对象中
                if(null != columnValue){
                    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(columnValue));
                }
            }
            // 4.4 将Put对象插入到表中
            table.put(put);
        }
        // 5. 如果是删除操作，则根据主键删除数据
        else if("delete".equals(operatorType)){
            // 5.1 获取键值
            String rowKeyValue = operatorData.getString(rowKeyColumn);
            // 5.2 创建Delete对象用于删除指定行的数据
            Delete delete = new Delete(Bytes.toBytes(rowKeyValue));
            // 5.3 删除指定列族
            table.delete(delete);
        }
        // 6.输出异常信息
        else {
            System.out.println("数据异常，操作类型：" + operatorType + "，HBase不支持");
        }
        // 关闭HBase连接
        table.close();
    }

    // 创建HBase表
    private void createHbaseTable(String tableName, String familyName) throws IOException {
        // 1.获取Admin对象
        Admin admin = connection.getAdmin();
        // 2.判断表是否存在
        TableName tab = TableName.valueOf(tableName);
        // 3.创建表
        if (!admin.tableExists(tab)) {
            // 创建表描述器
            TableDescriptorBuilder tabBuilder = TableDescriptorBuilder.newBuilder(tab);
            // 创建列族描述器
            ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(familyName));
            // 添加列族描述器
            tabBuilder.setColumnFamily(cfBuilder.build());
            // 创建表
            admin.createTable(tabBuilder.build());
        }
        // 4.关闭连接
        admin.close();
    }

    // 关闭连接
    @Override
    public void close() throws Exception {
        if(null != connection){
            connection.close();
        }
    }
}
